package com.gjy.kafka.resolve.consumer;

import com.gjy.kafka.resolve.config.KafkaConsumerConfig;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

/**
 * 手动提交 offset，业务成功后才提交
 *
 * @author gjy
 * @version 1.0
 * @since 2025-10-13 15:19:52
 */
public class ManualCommitConsumer implements AutoCloseable {

    private static final Logger logger = LoggerFactory.getLogger(ManualCommitConsumer.class);
    private final Consumer<String, String> consumer;

    public ManualCommitConsumer(String groupId) {
        Properties props = KafkaConsumerConfig.buildConfig(groupId);
        this.consumer = new KafkaConsumer<>(props);
    }

    /**
     * 订阅单个 topic 并循环消费
     */
    public void consume(String topic) {
        consumer.subscribe(Collections.singletonList(topic));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            if (records.isEmpty()) {
                continue;
            }
            for (ConsumerRecord<String, String> rec : records) {
                try {
                    // ====== 业务处理 ======
                    logger.info("消费到消息: topic={}, partition={}, offset={}, key={}, value={}",
                            rec.topic(), rec.partition(), rec.offset(), rec.key(), rec.value());

                    // 假设业务处理成功后才提交 offset
                    // commitSync 可保证在提交成功后才继续拉取下一批
                    consumer.commitSync(Collections.singletonMap(
                            new TopicPartition(rec.topic(), rec.partition()),
                            new OffsetAndMetadata(rec.offset() + 1)));
                } catch (Exception ex) {
                    // 业务处理异常：不提交 offset，保证下次重新消费
                    logger.error("业务处理异常，稍后重试: {}", ex.getMessage());
                    // 可选择 sleep 或者将记录写入错误队列
                }
            }
        }
    }

    @Override
    public void close() {
        consumer.wakeup(); // 触发退出 poll
        consumer.close();
    }

}
